package com.ruoyi.spark.controller;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.*;

import java.util.Properties;

@RestController
@Slf4j
public class SparkController {

   // @Autowired
    KafkaTemplate kafkaTemplate;

    static final String topic = "my_topic";
    @GetMapping("/send/{input}")
    public String sendTo(@PathVariable String input) {
        kafkaTemplate.send(topic, input);
        return "send success";
    }

    @KafkaListener(topics = topic)
    public void listener(String input) {
        if(input != null){
            System.out.println(input);
        }
        //log.info("input value：{}", input);
    }
    @RequestMapping("applog")
    public String sendToKafka(@RequestBody String logString) {
        log.info(logString);
        JSONObject obj = JSON.parseObject(logString);
        // 启动日志和事件日志进入到不同的 topic
        if (obj.getString("start") != null && obj.getString("start").length() > 0) {
            kafkaTemplate.send("gmall_startup_topic", logString);
        } else {
            kafkaTemplate.send("gmall_event_topic", logString);
        }
        return "ok";
    }
}
